package workflow

import (
	"fmt"
	"log"
	"sync"
	"time"

	"github.com/asynkron/protoactor-go/actor"
	"github.com/google/uuid"
	"github.com/louloulin/dataflare/pkg/actor/system"
	"github.com/louloulin/dataflare/pkg/workflow/connector"
)

// SourceActor is an actor that reads data from a source
type SourceActor struct {
	// id is the source ID
	id string
	// definition is the source definition
	definition SourceDefinition
	// instanceID is the workflow instance ID
	instanceID string
	// stateManager is the state manager PID
	stateManager *actor.PID
	// metricsCollector is the metrics collector PID
	metricsCollector *actor.PID
	// connector is the source connector
	connector connector.SourceConnector
	// targets are the target nodes
	targets map[string]*actor.PID
	// status is the node status
	status string
	// mu is a mutex for protecting the targets and status
	mu sync.RWMutex
	// logger is the source actor logger
	logger *log.Logger
	// running indicates if the source is running
	running bool
	// readCount is the number of records read
	readCount int64
	// errorCount is the number of errors
	errorCount int64
	// lastError is the last error
	lastError error
	// lastReadTime is the last time data was read
	lastReadTime time.Time
	// backpressure is the backpressure controller
	backpressure *BackpressureController
}

// BackpressureController controls backpressure
type BackpressureController struct {
	// maxCredits is the maximum number of credits
	maxCredits int
	// credits is the current number of credits
	credits int
	// mu is a mutex for protecting the credits
	mu sync.RWMutex
}

// NewBackpressureController creates a new backpressure controller
func NewBackpressureController(maxCredits int) *BackpressureController {
	return &BackpressureController{
		maxCredits: maxCredits,
		credits:    maxCredits,
	}
}

// TakeCredit takes a credit
func (b *BackpressureController) TakeCredit() bool {
	b.mu.Lock()
	defer b.mu.Unlock()

	if b.credits > 0 {
		b.credits--
		return true
	}
	return false
}

// ReturnCredit returns a credit
func (b *BackpressureController) ReturnCredit() {
	b.mu.Lock()
	defer b.mu.Unlock()

	if b.credits < b.maxCredits {
		b.credits++
	}
}

// AvailableCredits returns the number of available credits
func (b *BackpressureController) AvailableCredits() int {
	b.mu.RLock()
	defer b.mu.RUnlock()

	return b.credits
}

// NewSourceActor creates a new source actor
func NewSourceActor(id string, definition SourceDefinition, instanceID string, stateManager *actor.PID, metricsCollector *actor.PID) actor.Actor {
	return &SourceActor{
		id:               id,
		definition:       definition,
		instanceID:       instanceID,
		stateManager:     stateManager,
		metricsCollector: metricsCollector,
		targets:          make(map[string]*actor.PID),
		status:           "created",
		logger:           log.New(log.Writer(), fmt.Sprintf("[SourceActor-%s-%s] ", instanceID, id), log.LstdFlags),
		backpressure:     NewBackpressureController(100),
	}
}

// Receive handles messages for the source actor
func (s *SourceActor) Receive(context actor.Context) {
	switch msg := context.Message().(type) {
	case *actor.Started:
		s.handleStarted(context)
	case *actor.Stopping:
		s.handleStopping(context)
	case *ConnectNodeCommand:
		s.handleConnectNode(context, msg)
	case *StartNodeCommand:
		s.handleStartNode(context, msg)
	case *StopNodeCommand:
		s.handleStopNode(context)
	case *GetNodeStatusCommand:
		s.handleGetNodeStatus(context)
	case *CreditReturnedMessage:
		s.handleCreditReturned(context)
	case *ReadDataMessage:
		s.handleReadData(context)
	default:
		s.logger.Printf("Received unknown message: %T", msg)
	}
}

// handleStarted handles the Started message
func (s *SourceActor) handleStarted(context actor.Context) {
	s.logger.Println("Source actor started")

	// Create connector
	var err error
	s.connector, err = s.createConnector()
	if err != nil {
		s.logger.Printf("Failed to create connector: %v", err)
		s.status = "failed"
		context.Send(context.Parent(), &NodeErrorMessage{
			NodeID: s.id,
			Error:  fmt.Sprintf("Failed to create connector: %v", err),
		})
		return
	}

	// Report metrics
	if s.metricsCollector != nil {
		context.Send(s.metricsCollector, &system.ReportMetricCommand{
			Name:  "source.started",
			Value: 1,
			Tags: map[string]string{
				"source_id":   s.id,
				"instance_id": s.instanceID,
				"type":        s.definition.Type,
			},
		})
	}

	s.status = "initialized"
	s.logger.Println("Source actor initialized")
}

// handleStopping handles the Stopping message
func (s *SourceActor) handleStopping(context actor.Context) {
	s.logger.Println("Source actor stopping")

	// Stop connector
	if s.connector != nil {
		err := s.connector.Stop(context.Context())
		if err != nil {
			s.logger.Printf("Failed to stop connector: %v", err)
		}
	}

	// Report metrics
	if s.metricsCollector != nil {
		context.Send(s.metricsCollector, &system.ReportMetricCommand{
			Name:  "source.stopped",
			Value: 1,
			Tags: map[string]string{
				"source_id":   s.id,
				"instance_id": s.instanceID,
				"type":        s.definition.Type,
			},
		})
	}

	s.status = "stopped"
}

// handleConnectNode handles the ConnectNodeCommand message
func (s *SourceActor) handleConnectNode(context actor.Context, msg *ConnectNodeCommand) {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Add target
	s.targets[msg.TargetID] = msg.TargetPID

	// Respond with success
	context.Respond(&ConnectNodeResponse{
		Success: true,
	})

	s.logger.Printf("Connected to target node: %s", msg.TargetID)
}

// handleStartNode handles the StartNodeCommand message
func (s *SourceActor) handleStartNode(context actor.Context, msg *StartNodeCommand) {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Check if already running
	if s.running {
		context.Respond(&StartNodeResponse{
			Success: false,
			Error:   "source already running",
		})
		return
	}

	// Initialize connector
	err := s.connector.Init(context.Context(), s.definition.Config)
	if err != nil {
		s.logger.Printf("Failed to initialize connector: %v", err)
		context.Respond(&StartNodeResponse{
			Success: false,
			Error:   fmt.Sprintf("failed to initialize connector: %v", err),
		})
		return
	}

	// Start connector
	err = s.connector.Start(context.Context())
	if err != nil {
		s.logger.Printf("Failed to start connector: %v", err)
		context.Respond(&StartNodeResponse{
			Success: false,
			Error:   fmt.Sprintf("failed to start connector: %v", err),
		})
		return
	}

	// Start reading data
	s.running = true
	s.status = "running"
	s.lastReadTime = time.Now()

	// Schedule first read
	context.Send(context.Self(), &ReadDataMessage{})

	// Respond with success
	context.Respond(&StartNodeResponse{
		Success: true,
	})

	// Report status change
	context.Send(context.Parent(), &NodeStatusChangedMessage{
		NodeID: s.id,
		Status: "running",
	})

	s.logger.Println("Source started")
}

// handleStopNode handles the StopNodeCommand message
func (s *SourceActor) handleStopNode(context actor.Context) {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Check if running
	if !s.running {
		context.Respond(&StopNodeResponse{
			Success: false,
			Error:   "source not running",
		})
		return
	}

	// Stop connector
	err := s.connector.Stop(context.Context())
	if err != nil {
		s.logger.Printf("Failed to stop connector: %v", err)
		context.Respond(&StopNodeResponse{
			Success: false,
			Error:   fmt.Sprintf("failed to stop connector: %v", err),
		})
		return
	}

	// Update state
	s.running = false
	s.status = "stopped"

	// Respond with success
	context.Respond(&StopNodeResponse{
		Success: true,
	})

	// Report status change
	context.Send(context.Parent(), &NodeStatusChangedMessage{
		NodeID: s.id,
		Status: "stopped",
	})

	s.logger.Println("Source stopped")
}

// handleGetNodeStatus handles the GetNodeStatusCommand message
func (s *SourceActor) handleGetNodeStatus(context actor.Context) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	// Create metrics
	metrics := map[string]interface{}{
		"read_count":      s.readCount,
		"error_count":     s.errorCount,
		"last_read_time":  s.lastReadTime,
		"available_credits": s.backpressure.AvailableCredits(),
	}

	// Respond with status
	context.Respond(&GetNodeStatusResponse{
		Status:  s.status,
		Metrics: metrics,
		Error:   s.lastError,
	})
}

// handleCreditReturned handles the CreditReturnedMessage message
func (s *SourceActor) handleCreditReturned(context actor.Context) {
	// Return credit
	s.backpressure.ReturnCredit()

	// If we were blocked due to backpressure, try reading again
	if s.running && s.backpressure.AvailableCredits() == 1 {
		context.Send(context.Self(), &ReadDataMessage{})
	}
}

// handleReadData handles the ReadDataMessage message
func (s *SourceActor) handleReadData(context actor.Context) {
	// Check if running
	if !s.running {
		return
	}

	// Check if we have credits
	if !s.backpressure.TakeCredit() {
		s.logger.Println("Backpressure applied, waiting for credits")
		return
	}

	// Read data
	records, err := s.connector.Read(context.Context())
	if err != nil {
		s.logger.Printf("Failed to read data: %v", err)
		s.lastError = err
		s.errorCount++
		s.backpressure.ReturnCredit()

		// Report error metrics
		if s.metricsCollector != nil {
			context.Send(s.metricsCollector, &system.ReportMetricCommand{
				Name:  "source.errors",
				Value: 1,
				Tags: map[string]string{
					"source_id":   s.id,
					"instance_id": s.instanceID,
					"type":        s.definition.Type,
				},
			})
		}

		// Schedule next read
		time.AfterFunc(1*time.Second, func() {
			context.Send(context.Self(), &ReadDataMessage{})
		})
		return
	}

	// Update state
	s.lastReadTime = time.Now()
	if records != nil && len(records.Records) > 0 {
		s.readCount += int64(len(records.Records))

		// Send data to targets
		s.sendDataToTargets(context, records.Records)

		// Report metrics
		if s.metricsCollector != nil {
			context.Send(s.metricsCollector, &system.ReportMetricCommand{
				Name:  "source.records_read",
				Value: float64(len(records.Records)),
				Tags: map[string]string{
					"source_id":   s.id,
					"instance_id": s.instanceID,
					"type":        s.definition.Type,
				},
			})
		}
	} else {
		// No data, return credit immediately
		s.backpressure.ReturnCredit()
	}

	// Schedule next read
	if s.running {
		time.AfterFunc(100*time.Millisecond, func() {
			context.Send(context.Self(), &ReadDataMessage{})
		})
	}
}

// sendDataToTargets sends data to target nodes
func (s *SourceActor) sendDataToTargets(context actor.Context, records []connector.Record) {
	s.mu.RLock()
	defer s.mu.RUnlock()

	// Send data to each target
	for targetID, targetPID := range s.targets {
		for _, record := range records {
			// Create data message
			message := &system.DataMessage{
				MessageID: uuid.New().String(),
				Data:      record.Value,
				Metadata: map[string]string{
					"source_id":   s.id,
					"record_key":  string(record.Key),
					"instance_id": s.instanceID,
				},
				Timestamp:     time.Now(),
				SourceID:      s.id,
				DestinationID: targetID,
			}

			// Send message
			context.Send(targetPID, message)
		}
	}
}

// createConnector creates a source connector
func (s *SourceActor) createConnector() (connector.SourceConnector, error) {
	// Get connector factory
	factory := connector.GetSourceConnectorFactory(s.definition.Type)
	if factory == nil {
		return nil, fmt.Errorf("unknown source connector type: %s", s.definition.Type)
	}

	// Create connector
	return factory()
}

// StopNodeCommand is a command to stop a node
type StopNodeCommand struct{}

// StopNodeResponse is the response to a StopNodeCommand
type StopNodeResponse struct {
	// Success indicates if the operation was successful
	Success bool
	// Error is the error message if the operation failed
	Error string
}

// GetNodeStatusCommand is a command to get node status
type GetNodeStatusCommand struct{}

// GetNodeStatusResponse is the response to a GetNodeStatusCommand
type GetNodeStatusResponse struct {
	// Status is the node status
	Status string
	// Metrics are the node metrics
	Metrics map[string]interface{}
	// Error is the last error
	Error error
}

// CreditReturnedMessage is a message for credit return
type CreditReturnedMessage struct{}

// ReadDataMessage is a message for reading data
type ReadDataMessage struct{}
